
import data.DataPoint;
import extractor.SingleExtractor;
import fliter.SensorIdFliter;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.options.*;
import org.apache.beam.sdk.io.kafka.KafkaIO;
import org.apache.beam.sdk.io.hbase.HBaseIO;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.beam.sdk.transforms.*;
import org.apache.beam.sdk.transforms.windowing.*;
import org.apache.beam.sdk.values.KV;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.joda.time.Duration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import com.google.common.collect.ImmutableMap;
import outlierDetector.DynamicThresholdDetector;
import outlierDetector.StaticThresholdDetector;
import util.HbaseMap;
import util.MyTimestampFunction;
import util.PositiveMap;

/**
 * Created by emily on 17-11-27.
 */
public class OD {

    public interface OutliersDetectOptions extends PipelineOptions {

        @Description("kafka topic")
        @Default.String("Bridge")
        String getTopic();
        void setTopic(String topic);

    }

    public static void main(String[] args) {
        OutliersDetectOptions options = PipelineOptionsFactory.fromArgs(args).withValidation()
                .as(OutliersDetectOptions.class);
        Pipeline p = Pipeline.create(options);

        Configuration conf = null;
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "hadoop3,hadoop4,hadoop9");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        conf.set("hbase.master.port", "60000");
        conf.setLong("hbase.rpc.timeout", 60000);

        //Logger LOG = LoggerFactory.getLogger(OD.class);
        String sensor = "8AT002-DZ";

       // PCollection<KV<String,String>> lines =
                p.apply(KafkaIO.<String, String>read()
                .withBootstrapServers("hadoop7:9092, hadoop9:9092")
                .withTopic(sensor)
                .withKeyDeserializer(StringDeserializer.class)
                .withValueDeserializer(StringDeserializer.class)
                //.withMaxNumRecords(301)
                .withTimestampFn(new MyTimestampFunction())
                //.updateConsumerProperties(ImmutableMap.<String, Object>of("auto.offset.reset", "earliest"))
                .withoutMetadata()
        )


        //PCollection<DataPoint> PositiveDataStream = lines
                .apply("extractor", MapElements.via(new SingleExtractor()))
                .apply("sensorIdFliter", ParDo.of(new SensorIdFliter(sensor)))
                .apply("positiveMap", ParDo.of(new PositiveMap()))
        //.apply("Write", TextIO.write().to("bridge").withSuffix(".txt"));

        //PCollection<ArrayList<DataPoint>> thresholdStream = PositiveDataStream
                .apply(Window.<DataPoint>into(SlidingWindows.of(Duration.millis(1500)).every(Duration.millis(500)))
                .triggering(AfterPane.elementCountAtLeast(75))
                        .withAllowedLateness(Duration.millis(10))
                        .discardingFiredPanes())
                .apply("", Combine.globally(new DynamicThresholdDetector(75, 0.1)).withoutDefaults())
                .apply("", ParDo.of(new StaticThresholdDetector(1.0)))
                        .apply("", ParDo.of(new HbaseMap()))
                        .apply("write", HBaseIO.write()
                                .withConfiguration(conf)
                                .withTableId("Bridge_beam_outlier"));



        p.run().waitUntilFinish();


    }

}
